package com.atgugu.flink.chapter11.function;

import com.atgugu.flink.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

/**
 * @Author lzc
 * @Date 2022/4/12 13:36
 */
public class Flink02_Function_Table_1 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> stream = env.fromElements(
            new WaterSensor("hello agtuigu world", 1000L, 10),
            new WaterSensor("a b abc d", 2000L, 20),
            new WaterSensor("aaa cc dd", 3000L, 30),
            new WaterSensor("hello world abc d", 6000L, 60)
        );
    
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        Table table = tEnv.fromDataStream(stream);
        tEnv.createTemporaryView("sensor", table);
    
    
        // 自定义一个函数: 可以把小写字母变成大写字母
    
        // 1. 在table api中使用自定义函数
        // 1.1. 内联用法
        /*table
            // 左外连接
            .leftOuterJoinLateral(call(Split.class, $("id")))
            .select($("id"), $("word"), $("len"))
            .execute()
            .print();*/
    
        // 1.2. 先注册再使用
        
    
        // 2. 在sql中使用
        tEnv.createTemporaryFunction("my_split", Split.class);
        
        tEnv.sqlQuery("select" +
                          " id, " +
                          " word, " +
                          " len " +
                          "from sensor " +
                          "left outer join lateral table(my_split(id))  on true")
            .execute()
            .print();
       
    
    
    }
    
    
    @FunctionHint(output = @DataTypeHint("row<word string, len int>"))
    public static class Split extends TableFunction<Row> {
        public void eval(String s){
            if (s.contains("aaa")) {
                return;
            }
            
            
            String[] words = s.split(" ");
    
            for (String word : words) {
                // collect 每调用一次, 表示一行数据
                collect(Row.of(word, word.length()));
            }
            
        }
    }
    
    
}
/*
功能: 给你一个字符串, 把字符串展开办成多行, 每行有两列: 字符串本身  字符串的长度


"hello world atguigu"
        
        hello   5
        world   5
        atguigu 7
        
"hello atguigu"
         hello   5
         atguigu 7

 */
